package com.icarbonx.dap.notification.msgservice.rabbitmq

import java.util.concurrent.ConcurrentHashMap

import com.rabbitmq.client.AMQP.BasicProperties
import com.rabbitmq.client.{AMQP, ConnectionFactory, Connection, Channel}
;
import scala.collection.JavaConversions._
import com.rabbitmq.client.MessageProperties

/**
 * Created by Lenovo on 2018/8/9.
 */
object ConnUtil {
  @transient private lazy val pools: ConcurrentHashMap[MQObject, Connection] =
    new ConcurrentHashMap[MQObject, Connection]()

  def connect(mq: MQObject): Connection = {
    pools.getOrElseUpdate(mq, {
      val factory = new ConnectionFactory()
      factory.setHost(mq.host);
      factory.setPort(mq.port);
      factory.setUsername(mq.username);
      factory.setPassword(mq.password);
      factory.setVirtualHost(mq.virtualhost);
      // 通过工厂获取连接
      factory.newConnection()
    }
    )
  }

  def close(mq: MQObject): Unit = {
    pools.get(mq).close()
  }

  val hook = new Thread {
    override def run = pools.values().foreach(_.close())
  }
  sys.addShutdownHook(hook.run)

  //  def sendMsg(mqChannel: Channel, exchangeName: String, mqRouteKey: String, msg: String): Unit = {
  //    // 声明exchange
  //    mqChannel.exchangeDeclare(exchangeName, "topic")
  //    // 发送消息, 指定RoutingKey
  //    mqChannel.basicPublish(exchangeName, mqRouteKey,
  //      new AMQP.BasicProperties.Builder().contentType("application/json").deliveryMode(2).`type`("String").build(), msg.getBytes("UTF-8"))
  //    //    mqChannel.basicPublish(exchangeName, mqRouteKey, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.toString.getBytes("UTF-8"))
  //  }

  def sendMsg(mqChannel: Channel, exchangeName: String, mqRouteKey: String, msg: String): Unit = {
    // 声明exchange
    mqChannel.exchangeDeclare(exchangeName, "topic", true)
    // 发送消息, 指定RoutingKey
    mqChannel.basicPublish(
      exchangeName,
      mqRouteKey,
      null,
      msg.toString.getBytes("UTF-8")
    )
    println("____ msg :" + msg)
  }

}
